# coding=utf-8
import os
import pandas as pd
import fastparquet

from core.m_logging import logger


class ParquetFileOutput(object):
    def __init__(self, file_name='df.gzip.parquet', flush_size=1000):
        self.dict = {}
        self.file_name = file_name
        self.flush_size = flush_size
        self.length = 0

    def cache(self, kv_pairs):
        logger.info("caching kv_pairs...")
        for column, value in kv_pairs.iteritems():
            if column not in self.dict:
                self.dict[column] = [value]
            else:
                self.dict[column].append(value)
        self.length += 1
        logger.info("kv_pairs is cached, current length is %s, kv_pairs: omitted" % self.length)
        if self.length >= self.flush_size:
            self.flush()

    def flush(self):
        logger.info("flushing parquet to file...")
        df_to_append = pd.DataFrame(self.dict)
        if os.path.exists(self.file_name):
            df_from_disk = pd.read_parquet(self.file_name)
            df_to_disk = df_from_disk.append(df_to_append, ignore_index=True)
        else:
            df_to_disk = df_to_append
        logger.info("writing to %s..." % self.file_name)
        df_to_disk.to_parquet(self.file_name, compression='gzip')
        self.dict = {}
        self.length = 0

    def output(self, content):
        logger.info('output file to pqrquet file %s...' % self.file_name)
        if "ansiList" in content:
            for asin in content["ansiList"]:
                d = {"asin": str(asin["asin"]),
                     "summary_url": str(asin["product_summary_url"]),
                     "review_url": str(asin["review_url"]),
                     "product_name": str(asin["product_name"]),
                     "product_image_url": str(asin["product_image_url"]),
                     "product_rating": str(asin["product_rating"]),
                     "customer_review": str(asin["customer_review"]),
                     "product_price": str(asin["product_price"]),
                     "shipping": str(asin["shipping"]),
                     "stock": str(asin["stock"]),
                     "choice_type": str(asin["choice"]["choice_type"]),
                     "choice_count": str(asin["choice"]["choice_count"]),
                     "choice_price": str(asin["choice"]["choice_price"]),
                     "best_seller": str(asin["best_seller"]),
                     "node_no": str(content["node_no"]),
                     "page_no": str(content["page_no"]),
                     "scrape_date": str(content["scrape_date"])}
                self.cache(d)
        else:
            logger.error("ansiList not in content")
